package net.bwie.realtime.jtp.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.JdbcUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

/**
 * 简介说明:
 *
 * 将APP流量日志保存到clickhouse表工具类
 * @author: LiLi
 * @date: 2025/05/22 14:08:18
 * @version: 1.0
 */
public class JtpLogClickHouse {
    public static void savePageLog (DataStream<String> stream){
//        数据转换
        SingleOutputStreamOperator<String> jsonstream = stream.map(new RichMapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                JSONObject object = JSON.parseObject(s);
                JSONObject commonObject = object.getJSONObject("common");
                JSONObject pageObject = object.getJSONObject("page");


                // 2提前字段值
                JSONObject jsonObject = new JSONObject() ;
                jsonObject.put("mid", commonObject.getString("mid")) ;
                jsonObject.put("area_code", commonObject.getString("ar")) ;
                jsonObject.put("brand", commonObject.getString("ba")) ;
                jsonObject.put("channel", commonObject.getString("ch")) ;
                jsonObject.put("is_new", commonObject.getString("is_new")) ;
                jsonObject.put("model", commonObject.getString("md")) ;
                jsonObject.put("os", commonObject.getString("os")) ;
                jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString()) ;
                jsonObject.put("version_code", commonObject.getString("vc")) ;
                jsonObject.put("page_id", pageObject.getOrDefault("page_id", "")) ;
                jsonObject.put("last_page_id", pageObject.getOrDefault("last_page_id", "")) ;
                jsonObject.put("during_time", pageObject.getLong("during_time")) ;
                jsonObject.put("sourceType", pageObject.getOrDefault("source_type", "")) ;
                jsonObject.put("item_type", pageObject.getOrDefault("item_type", "")) ;
                jsonObject.put("item", pageObject.getOrDefault("item", "")) ;
                jsonObject.put("ts", object.getLong("ts")) ;
                // 3返回JSON字符串
                return jsonObject.toJSONString();

            }
        });

        // 2-1字段名称
        String[] columns = new String[]{
                "mid", "area_code","brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "page_id", "last_page_id", "during_time", "sourceType", "item_type", "item", "ts"
        };
        // 2-2插入语句
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_app_page_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    page_id, last_page_id, during_time, sourceType, item_type, item,\n" +
                "    ts\n" +
                ") VALUES (\n" +
                "   ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                ")" ;
        // 2-3插入数据
        JdbcUtil.sinkToClickhouseUpsertJson(jsonstream,columns, upsertSql);
    }

    public static void saveStartLog(DataStream<String> stream) {
    /*
       {
          "common": {
             "ar": "230000",
             "uid": "57",
             "os": "iOS 13.3.1",
             "ch": "Appstore",
             "is_new": "0",
             "md": "iPhone Xs",
             "mid": "mid_495168",
             "vc": "v2.1.134",
             "ba": "iPhone"
          },
          "page": {
             "page_id": "good_detail",
             "item": "4",
             "during_time": 19709,
             "item_type": "sku_id",
             "last_page_id": "good_list",
             "source_type": "query"
          },
          "ts": 1736784135000
       }
     */
        // 1数据转换
        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        // 1解析JSON
                        JSONObject object = JSON.parseObject(value);
                        JSONObject commonObject = object.getJSONObject("common");
                        JSONObject startObject = object.getJSONObject("start");

                        // 2提前字段值
                        JSONObject jsonObject = new JSONObject() ;
                        jsonObject.put("mid", commonObject.getString("mid")) ;
                        jsonObject.put("area_code", commonObject.getString("ar")) ;
                        jsonObject.put("brand", commonObject.getString("ba")) ;
                        jsonObject.put("channel", commonObject.getString("ch")) ;
                        jsonObject.put("is_new", commonObject.getString("is_new")) ;
                        jsonObject.put("model", commonObject.getString("md")) ;
                        jsonObject.put("os", commonObject.getString("os")) ;
                        jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString()) ;
                        jsonObject.put("version_code", commonObject.getString("vc")) ;

                        jsonObject.put("entry", startObject.getString("entry")) ;
                        jsonObject.put("loading_time", startObject.getLong("loading_time")) ;
                        jsonObject.put("open_ad_id", startObject.getLong("open_ad_id")) ;
                        jsonObject.put("open_ad_ms", startObject.getLong("open_ad_ms")) ;
                        jsonObject.put("open_ad_skip_ms", startObject.getLong("open_ad_skip_ms")) ;

                        jsonObject.put("ts", object.getLong("ts")) ;
                        // 3返回JSON字符串
                        return jsonObject.toJSONString();
                    }
                }
        );

        // 2-1字段名称
        String[] columns = new String[]{
                "mid", "area_code","brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "entry", "loading_time", "open_ad_id", "open_ad_ms", "open_ad_skip_ms",  "ts"
        };
        // 2-2插入语句
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_app_start_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    entry, loading_time, open_ad_id, open_ad_ms, open_ad_skip_ms,\n" +
                "    ts\n" +
                ") VALUES (\n" +
                "     ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                " )" ;
        // 2-3插入数据
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream,columns, upsertSql);
    }

}
